package ru.mipt.victator.pagerank;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.Properties;

import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.FloatWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;

public class PreProcessing
{
	private FileSystem FS;
	private JobConf conf;
	private Path pText;
	private Path pList;
	private float StartPR;
	
	public void config(FileSystem _FS, Properties _props)
	{
		FS = _FS;
		conf = new JobConf();
		pText = new Path(_props.getProperty("InputData.PageList.File"));
		pList = new Path(_props.getProperty("OutputData.PageRank.Dir"),"0");
		StartPR = Float.parseFloat(_props.getProperty("InputData.StartPageRank.Float"));
	}
	
	public int process() throws IOException
	{
		int c=0;
		//input
		FSDataInputStream FSdis = FS.open(pText);
		BufferedReader br = new BufferedReader(
												new InputStreamReader(FSdis,"UTF-8")
												);
		String in;
		
		//output
		SequenceFile.Writer writer = SequenceFile.createWriter(	FS,
																conf,
																pList,
																Text.class,
																FloatWritable.class
																);
		//process
		while((in = br.readLine()) != null)
		{
			c++;
			writer.append(new Text(in), new FloatWritable(StartPR));
		}
		writer.close();
		return c;
	}
}
